/**
 * 
 */
package com.inspur.cloud.demo.service.flinksql;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.inspur.cloud.demo.entity.FlinkConfig;
import com.inspur.cloud.demo.entity.Zeppelin;
import com.inspur.cloud.demo.repository.FlinkRespository;
import com.inspur.cloud.demo.repository.ZeppelinRespository;
import com.inspur.cloud.demo.util.HttpUtils;
import com.inspur.cloud.demo.util.Response;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author yuanfq
 * @version 1.0
 * 2020年3月25日 下午3:47:31<br>
 */
@Service
public class FlinkServiceImpl implements FlinkSqlService{
    private FlinkConfig flinkConfig ;
    private String zeppelinUrl;
    private Zeppelin zeppelin;
    @Autowired
    ZeppelinRespository zeppelinRespository;
    @Autowired
    public FlinkServiceImpl(FlinkRespository flinkRepository,ZeppelinRespository zeppelinRespository) throws Exception{
        this.flinkConfig = flinkRepository.findByType("dev");
        this.zeppelinUrl = flinkConfig.getZeppelinserver()+":"+flinkConfig.getZeppelinport();
        this.zeppelin = zeppelinRespository.findByVersion("1.0");
        //如果取不到 那么需要新创建一个
        if(zeppelin==null)
        {
            createDynamicSql(zeppelinRespository);
            this.zeppelin = zeppelinRespository.findByVersion("1.0");
        }
    }
    
    @Override
    public FlinkConfig getFlinkConfig() {
        // TODO Auto-generated method stub
        return flinkConfig;
    }
    @Override
    public Response<?> runSql(String notebookid,String paragid,Map<String,Object> param) throws Exception {
        String retrunstr = HttpUtils.doPost(zeppelinUrl + "/api/notebook/run/"+ notebookid+"/"+ paragid, param,"utf-8");
        JSONObject respoJson =  (JSONObject)JSONObject.parse(retrunstr);
        String returntype ="table";//默认是table
        Map resultMap = new HashMap();


        if(respoJson.get("status").equals("INTERNAL_SERVER_ERROR"))//发生错误的情况
        {
            returntype="error";
            analyDefaultJsonObject(respoJson,resultMap);

        }else {
            Map body = (Map) respoJson.get("body");//里面有两个数据 msg  code
            JSONArray msg = (JSONArray) body.get("msg");//里面有两个数据 data 和type
            //循环
            for (int i = 0; i < msg.size(); i++) {
                JSONObject dataInfo = msg.getJSONObject(i);
                if (dataInfo.get("type").equals("TABLE")) {
                    analyTableJsonObject(dataInfo, resultMap);
                } else if (dataInfo.get("type").equals("TEXT")) {
                    returntype = "text";
                    analyTextJsonObject(dataInfo, resultMap);
                } else {

                }
            }
        }
        resultMap.put("returntype",returntype);
        return Response.success(resultMap);
    }
    @Override
    public Response<?> addParagraph(String notebookid,Map<String, Object> param)  throws Exception
    {
        Map resultMap = new HashMap();
        String retrunstr = HttpUtils.doPost(zeppelinUrl + "/api/notebook/"+ notebookid+"/paragraph ", param,"utf-8");
        JSONObject respoJson =  (JSONObject)JSONObject.parse(retrunstr);
        if(respoJson.get("status").equals("CREATED"))
        {
            resultMap.put("returnuStatus","OK");
            resultMap.put("returnuMessage",respoJson.get("body"));

        }else
        {
            resultMap.put("returnuStatus","error");
            resultMap.put("returnMessage",respoJson.get("message"));

        }
        return Response.success(resultMap);
    }

    //更新保存一个段落
    @Override
    public Response<?> saveParagraph(String notebookid,String paragid,Map<String, Object> param)  throws Exception
    {
        Map resultMap = new HashMap();

        String retrunstr = HttpUtils.doPut(zeppelinUrl + "/api/notebook/"+ notebookid+"/paragraph/+ "+paragid, param,"utf-8");
        JSONObject respoJson =  (JSONObject)JSONObject.parse(retrunstr);
        if(respoJson.get("status").equals("OK"))
        {
            resultMap.put("returnuStatus","OK");

        }else
        {
            resultMap.put("returnuStatus","error");
            resultMap.put("returnMessage",respoJson.get("message"));

        }

        return Response.success(resultMap);
    }
    //删除一个段落
    @Override
    public Response<?> deleteParagraph(String notebookid,String paragid)  throws Exception
    {
        Map resultMap = new HashMap();
        String retrunstr = HttpUtils.doDelete(zeppelinUrl + "/api/notebook/"+ notebookid+"/paragraph/+ "+paragid);
        JSONObject respoJson =  (JSONObject)JSONObject.parse(retrunstr);
        if(respoJson.get("status").equals("OK"))
        {
            resultMap.put("returnuStatus","OK");

        }else
        {
            resultMap.put("returnuStatus","error");
            resultMap.put("returnMessage",respoJson.get("message"));

        }

        return Response.success(resultMap);
    }

    public boolean checkParagraph() throws Exception
    {
        String retrunstr = HttpUtils.doGet(zeppelinUrl + "/api/notebook/"+ zeppelin.getNoteid()+"/paragraph/"+ zeppelin.getParagraphsid());

        JSONObject respoJson =  (JSONObject)JSONObject.parse(retrunstr);

        if(respoJson.get("status").equals("NOT_FOUND"))
        {
            //创建
            if(respoJson.get("message").equals("No such paragraph: paragraph not found"))
            {
                createDynamicSql(zeppelinRespository);
                this.zeppelin = zeppelinRespository.findByVersion("1.0");
                return true;
            }else if (respoJson.get("message").equals("No such note: note not found"))
            {
                createDynamicSql(zeppelinRespository);//后期考虑不重新生成note
                this.zeppelin = zeppelinRespository.findByVersion("1.0");
                return true;
            }
            return false;
        }

        return true;
    }

    @Override
    public Response<?> runSql(Map<String,Object> param) throws Exception {
        //可以选择版本开发 或者测试 这个在配置文件类胚子
        //取最新的一个
        //检测一下是否可用 如果不可用则 需要重新创建
        if(!checkParagraph())
        {
            return Response.fail();
        }
        String retrunstr = HttpUtils.doPost(zeppelinUrl + "/api/notebook/run/"+ zeppelin.getNoteid()+"/"+ zeppelin.getParagraphsid(), param,"utf-8");
        JSONObject respoJson =  (JSONObject)JSONObject.parse(retrunstr);
        Map body = (Map)respoJson.get("body");//里面有两个数据 msg  code
        JSONArray msg = (JSONArray)body.get("msg");//里面有两个数据 data 和type
        String returntype ="table";//默认是table
        Map resultMap = new HashMap();
        for(int i=0;i<msg.size();i++){
            //3、把里面的对象转化为JSONObject
            JSONObject dataInfo = msg.getJSONObject(i);
            // {"status":"OK","body":{"code":"SUCCESS","msg":[{"type":"TABLE","data":"access\tdata\ngll\t34\nmsp\t400\nuer\t100,0\nzs\t300.12\n"}]}}
            if(dataInfo.get("type").equals("TABLE"))
            {
                analyTableJsonObject(dataInfo,resultMap);
            }else if(dataInfo.get("type").equals("TEXT"))
            {
                returntype="text";
                analyTextJsonObject(dataInfo,resultMap);
            }
        }

        resultMap.put("type",returntype);

        return Response.success(resultMap);
    }

    public void analyDefaultJsonObject(JSONObject dataInfo,Map resultMap)
    {
        String dataStr = ((String) dataInfo.get("INTERNAL_SERVER_ERROR"));
        resultMap.put("data",dataStr);
    }

    public void analyTextJsonObject(JSONObject dataInfo,Map resultMap)
    {
        String dataStr = ((String) dataInfo.get("data"));
        resultMap.put("data",dataStr);
    }

    public void analyTableJsonObject(JSONObject dataInfo,Map resultMap)
    {
        List<Map> columnList = new ArrayList<>();//表头
        List<Map> columnDataList = new ArrayList<>();//数据
        int num =0;

        //开始解析string  先\n 解析第一行是表头 后面是数据
        String dataStr[] = ((String) dataInfo.get("data")).split("\\n");
        String columnHead[]=null;
        for(String tableInfo :dataStr) {
            //取表头
            if (num == 0) {
                columnHead = tableInfo.split("\\t");
                for (String columnHeadstr : columnHead) {
                    HashMap<String, Object> columnInfo = new HashMap<>();
                    columnInfo.put("field", columnHeadstr);
                    columnInfo.put("title", columnHeadstr);
                    columnInfo.put("align", "center");
                    columnList.add(columnInfo);
                }
            } else//取数据
            {
                String columnDataStr[] = tableInfo.split("\\t");
                Map columnData = new HashMap<>();
                for (int datai = 0; datai < columnDataStr.length; datai++) {
                    columnData.put(columnHead[datai], columnDataStr[datai]);
                }
                columnDataList.add(columnData);
            }
            num++;
        }
        resultMap.put("data",columnDataList);
        resultMap.put("columndata",columnList);
        resultMap.put("count",columnDataList.size());

    }


    public  void  createDynamicSql(ZeppelinRespository zeppelinRespository) throws Exception{
        Map<String,Object> params = new HashMap<>();
        params.put("name","flinkDynamicSql勿删");
        //首先查询一下名称是否存在
        String noteIdstr = HttpUtils.doGet(zeppelinUrl+"/api/notebook");
        JSONObject jsonObject = JSONObject.parseObject(noteIdstr);
        JSONArray notebookArray = (JSONArray)jsonObject.get("body");
        //循环查找
        JSONObject notepadObject;
        for(int i =0;i<notebookArray.size();i++){
            notepadObject = (JSONObject)notebookArray.get(i);
            // 删除重建
            if(StringUtils.compare((String)notepadObject.get("path"),"/flinkDynamicSql勿删")==0)
            {
                HttpUtils.doDelete(zeppelinUrl+"/api/notebook/"+notepadObject.get("id"));
                break;
            }
        }

        noteIdstr = HttpUtils.doPostJson(zeppelinUrl+"/api/notebook", params,"utf-8");
        jsonObject = JSONObject.parseObject(noteIdstr);
        Zeppelin zeppelin = new Zeppelin();
        zeppelin.setNoteid((String)jsonObject.get("body"));
        //取到notepadid 以后创建 paragraphs 输入自动化地址信息
        params.clear();
        params.put("title","run sql勿删");
        params.put("text","%flink.ssql(type=update)\n${sql}");
        String paragraphsIdstr= HttpUtils.doPostJson(zeppelinUrl +"/api/notebook/"+(String)jsonObject.get("body")+"/paragraph", params,"utf-8");
        jsonObject = JSONObject.parseObject(paragraphsIdstr);
        zeppelin.setDescription("测试");
        zeppelin.setOrderseq(0);
        zeppelin.setParagraphsid((String)jsonObject.get("body"));
        zeppelin.setVersion("1.0");
        zeppelinRespository.deleteAll();
        zeppelinRespository.saveAndFlush(zeppelin);
    }
    @Override
    public  Response<?> createDynamicSql() throws Exception{
        //首先创建一个note
        createDynamicSql(zeppelinRespository);
        return Response.success();
    }
    @Override
    public Response<?> queryGroupJobSql()
    {
        String noteIdstr = HttpUtils.doGet(zeppelinUrl+"/api/notebook");
        JSONObject jsonObject = JSONObject.parseObject(noteIdstr);
        JSONArray notebookArray = (JSONArray)jsonObject.get("body");
        //对数据进行分组排序
        //return Response.success(analyGroup(notebookArray));
        return Response.success(analyGroupRec(notebookArray));
    }

    //普通写法  支持到 两级
    public ArrayList<Map<String,Object>> analyGroup(JSONArray notebookArray)
    {
        ArrayList<Map<String,Object>> returnArrayList = new ArrayList<>();
        Map<String,Object> jobMap=null;
        String tmpGroupname="";
        ArrayList<Map<String,Object>> tmpArrayList = new ArrayList<>();
        JSONObject jsonObject =null;
        ArrayList<String> pathList = new ArrayList<>();
        
        for(int i =0 ;i<notebookArray.size();i++)
        {
            jsonObject = (JSONObject)notebookArray.get(i);//包含 path / 以及 id
            String path[] = ((String)jsonObject.get("path")).split("/");
            pathList.clear();
            //从1开始去掉前面的空串
            for (int p=1;p<path.length;p++)
            {
                pathList.add(path[p]);
            }
            String tmpname = pathList.get(0);
            //控制是否新生产map 以及 arrlist
            if(tmpGroupname.equals("")) {//首次进入
                tmpGroupname = tmpname;
                jobMap = new HashMap<>();
                jobMap.put("id", jsonObject.get("id")); //第一个的id
                jobMap.put("name", tmpname);  // 组名称
                jobMap.put("zeppelinUrl",zeppelinUrl);
                jobMap.put("isGroup","false");
            }else if(!tmpname.equals(tmpGroupname)) {
                returnArrayList.add(jobMap);
                jobMap = new HashMap<>();
                //分组信息已经改变 需要 重新声明一个
                tmpArrayList = new ArrayList<>();
                tmpGroupname = tmpname;
                jobMap.put("id", jsonObject.get("id")); //第一个的id
                jobMap.put("name", tmpname);  // 组名称
                jobMap.put("zeppelinUrl",zeppelinUrl);
                jobMap.put("isGroup","false");
            }
            if(path.length>=3) {
                Map<String, Object> subInfo = new HashMap<>();
                subInfo.put("name", path[2]);
                subInfo.put("id", jsonObject.get("id"));
                tmpArrayList.add(subInfo);
                jobMap.put("childrenList", tmpArrayList);
                jobMap.put("isGroup","true");
            }

        }
        return returnArrayList;
    }


    //递归一定要是一个整体 传导
    //数组 处理的步骤
    public ArrayList<Map<String,Object>> analyGroupRec(JSONArray notebookArray)
    {
        ArrayList<Map<String,Object>> returnList = new ArrayList<>();
        String tmpName ="";
        ArrayList<Map<String,Object>> subGroupList = null;

        for(int i =0 ;i<notebookArray.size();i++) {
            JSONObject jsonObject = (JSONObject)notebookArray.get(i);
            String pathArray[] = ((String)jsonObject.get("path")).split("/");
            String id = (String)jsonObject.get("id");
            HashMap<String,Object> itmeMap =null;
            //第一次 或者 本次的名称 不等于上次的名称 也就是新的分组进来
            if(i==0 || !tmpName.equals(pathArray[1]))
            {
                tmpName = pathArray[1];  //更新临时变量的 分组名称
                itmeMap = new HashMap<>();
                itmeMap.put("name",pathArray[1]);
                itmeMap.put("zeppelinUrl",zeppelinUrl);
                itmeMap.put("id",id);
                if(pathArray.length>=3) {  //如果 路径超过两个  那么说明有子节点 先新增一个数组 然后去添加子节点
                    subGroupList = new ArrayList<Map<String,Object>>();
                    addSubItem(pathArray,2,id,subGroupList);  //把数组  returnList中包含父节点的名称 以及 新增子节点的list
                    itmeMap.put("childrenList",subGroupList);
                    returnList.add(itmeMap);

                }else
                {
                    returnList.add(itmeMap);
                }
            }else if(tmpName.equals(pathArray[1]) )//关键就是这个地方
            {
                addSubItem(pathArray,2,id,subGroupList);
            }
        }
        return returnList;
    }

    //pathArray
    public void addSubItem(String pathArray[],int startIndex,String id,ArrayList<Map<String,Object>> subGroupList)
    {
        HashMap<String, Object> itmeSubMap =null;
        //取最后一个对比
        //当第一次的时候 有0 个 后面的话 有多0个的时候需要新增
        Map<String, Object> tmpMap =null;
        if(subGroupList.size()>0) {
            tmpMap = subGroupList.get(subGroupList.size() - 1);
        }else
        {
            tmpMap = new HashMap<>();
        }
        //先实例一下节点
        itmeSubMap = new HashMap<>();
        itmeSubMap.put("parentName", pathArray[startIndex-1]);
        itmeSubMap.put("name", pathArray[startIndex]);
        itmeSubMap.put("id", id);

        //说明有子节点已经初始化至少有一个
        if(tmpMap.size()>0 ){
            //如果最后这个节点名称现在的节点名称一样  那么说明如果没有下一级
            if(tmpMap.get("name").equals(pathArray[startIndex]))
            {
                // 如果这个子节点还有 子节点 那么继续递归
                if(pathArray.length-1 > startIndex) {
                    ArrayList<Map<String,Object>> tmpArrayList = (ArrayList<Map<String,Object>>)tmpMap.get("childrenList");
                    if(tmpArrayList==null)
                    {
                        tmpArrayList= new ArrayList<Map<String,Object>>();
                        itmeSubMap.put("childrenList",tmpArrayList);
                        //如果中间路径相同，不能重复添加  比如  放在这 可以判断 是否已经添加过了 名称相同 如果
                        //2020-7-10晚 终于搞定了
                        subGroupList.add(itmeSubMap);
                    }
                    addSubItem(pathArray, startIndex + 1, id, tmpArrayList);
                }
                //先添加这个节点  最底上加，不然会出现重复 上面的递归也
                //System.out.println("添加子节点"+pathArray[startIndex]);
                //没有子节点 说明这是个叶子跟 前面的 文件夹同名
                if(pathArray.length-1 == startIndex) {
                    subGroupList.add(itmeSubMap);
                }
            }else  //如果不一样 两种情况 处理  最终节点的话 添加 如果不是最终节点 那需要 分支了
            {
                //到了最下面 名称不一样 但是已经到里 最底层
                if(pathArray.length-1 == startIndex) {
                    subGroupList.add(itmeSubMap);
                }else  if(pathArray.length-1 > startIndex) {  //后面还有子节点的话
                    ArrayList<Map<String,Object>> tmpArrayList =new ArrayList<>();
                    itmeSubMap.put("childrenList",tmpArrayList);
                    subGroupList.add(itmeSubMap);
                    addSubItem(pathArray, startIndex + 1, id, new ArrayList<>());
                }
            }
        }else//如果是首个子节点
        {
            // 如果还没到底 减一 去掉最前面的空格
            if(pathArray.length-1 > startIndex) {
                ArrayList<Map<String,Object>> tmpArrayList = (ArrayList<Map<String,Object>>)tmpMap.get("childrenList");
                if(tmpArrayList==null)
                {
                    tmpArrayList = new ArrayList<>();
                }
                itmeSubMap.put("childrenList",tmpArrayList);
                addSubItem(pathArray, startIndex + 1, id, tmpArrayList);
            }
            //如果子节点
            subGroupList.add(itmeSubMap);
        }
    }
    //查询所有子任务组名称
    @Override
    public Response<?> querySubGroupJobSql(Map<String,Object> param){
        String id = (String)param.get("id");
        String noteIdstr = HttpUtils.doGet(zeppelinUrl+"/api/notebook/"+id);
        JSONObject jsonObject = JSONObject.parseObject(noteIdstr);
        //对数据进行分组排序
        return Response.success(jsonObject.get("body"));
    }


}
